/*
 * Copyright (c) 2020 - present, Inspur Genersoft Co., Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.iec.edp.caf.lock.service.redis;

import io.iec.edp.caf.common.JSONSerializer;
import io.iec.edp.caf.commons.exception.ExceptionLevel;

import io.iec.edp.caf.commons.entity.CAFContextInfo;
import io.iec.edp.caf.commons.spi.utils.CAFContextResolverUtils;
import io.iec.edp.caf.commons.utils.StringUtils;
import io.iec.edp.caf.lock.service.api.api.*;
import io.iec.edp.caf.lockservice.api.LockStatus;
import io.iec.edp.caf.lock.service.api.exception.DLErrorDefinition;

import io.iec.edp.caf.lock.service.api.utils.DataValidator;
import io.iec.edp.caf.lockservice.api.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.lang.Nullable;


import java.io.IOException;
import java.time.Duration;
import java.util.*;
import java.util.stream.Collectors;


/**
 * 数据锁服务实现类
 *
 * @author wangyandong
 */
@Slf4j
public class DistributedLockServiceImpl implements LockService {

    private DistributedLockFactory factory;


    private LockExtension lockExtension;



    public DistributedLockServiceImpl(DistributedLockFactory factory,LockExtension lockExtension) {
        this.factory = factory;
        this.lockExtension=lockExtension;
    }

    @Override
    public LockResult addLock(String mkId, String dataCat, String dataId, DataLockOptions option, String funcId, String comment, @Nullable DistributedLockOptions distributedLockOptions) {
        DataValidator.checkForEmptyString(mkId, "mkId");
        DataValidator.checkForEmptyString(dataCat, "dataCat");
        DataValidator.checkForEmptyString(dataId, "dataId");
        DataValidator.checkForNullReference(option, "options");
        LockResult result = this.internalAddLock(mkId, dataCat, dataId, option, funcId, comment, null, distributedLockOptions);

        return result;
    }

    /**
     * @param mkId    模块ID
     * @param dataCat 数据种类
     * @param dataId  数据ID
     * @param option  数据锁的设置选项
     * @param funcId  加锁的功能Id
     * @param comment 备注
     * @return
     */
    @Override
    public LockResult addLock(String mkId, String dataCat, String dataId, DataLockOptions option, String funcId, String comment) {
        return this.addLock(mkId, dataCat, dataId, option, funcId, comment, null);
    }

    @Override
    @Deprecated
    public BatchLockResult addBatchLock(String mkId, String dataCat, List<String> dataIds, String groupId, DataLockOptions option, String funcId, String comment, @Nullable DistributedLockOptions distributedLockOptions) {
        return this.addBatchLock(mkId, dataCat, dataIds, groupId, option.getPersistenceTime(), funcId, comment);
    }

    @Override
    @Deprecated
    public BatchLockResult addBatchLock(String mkId, String dataCat, List<String> dataIds, String groupId, DataLockOptions option, String funcId, String comment) {
        return this.addBatchLock(mkId, dataCat, dataIds, groupId, option, funcId, comment, null);
    }

    /**
     * 批量锁加锁方法
     *
     * @param mkId            模块ID
     * @param dataCat         数据种类
     * @param dataIds         数据ID列表
     * @param groupId         批量锁标识，可空
     * @param persistenceTime 持续时间
     * @param funcId          功能ID
     * @param comment         备注
     * @return
     */
    @Override
    public BatchLockResult addBatchLock(String mkId, String dataCat, List<String> dataIds, @Nullable String groupId, Duration persistenceTime, String funcId, String comment) {
        DataValidator.checkForEmptyString(mkId, "mkId");
        DataValidator.checkForEmptyString(dataCat, "dataCat");
        DataValidator.checkForNullReference(dataIds, "dataIDs");
        DataValidator.checkForNullReference(persistenceTime, "persistenceTime");
//        if (isSessionNotValid(this.contextService.getSessionId())) {
//            throw new io.iec.edp.caf.lockservice.api.DistributedLockException("pfcomm", "Gsp_Svc_DistributedLock_1002", "Session:" + this.contextService
//                    .getSessionId() + " is not online!", null, ExceptionLevel.Error, false);
//        }

        if (StringUtils.isEmpty(groupId)) {
            groupId = UUID.randomUUID().toString();
        }
        BatchLockResult result = new BatchLockResult();
        result.setGroupId(groupId);
        if(dataIds==null||dataIds.size()==0){
            result.setSuccess(true);
            return result;
        }
        Set<String> set = new HashSet<>(dataIds);
        if(log.isDebugEnabled()){
            log.debug("dataIds length:"+dataIds.size());
            log.debug("dataSet length:"+set.size());
        }
        dataIds = set.stream().collect(Collectors.toList());
        if(log.isDebugEnabled()){
            log.debug("filterDataIds length:"+dataIds.size());
            log.debug("dataIds :"+ JSONSerializer.serialize(dataIds));
        }
        DistributedLock distributedLock=null;
        //根据上下文获取租户ID
        CAFContextInfo contextInfo= CAFContextResolverUtils.getCAFContextInfo();
        int tenantId =0;
        if(contextInfo!=null){
            tenantId=contextInfo.getTenantId();
        }
        String resourcePrefx= String.format("%s-%s-%s", tenantId, mkId, dataCat);
        Duration expiryTime = Duration.ofSeconds(30) ;
        try{
            long time1=System.currentTimeMillis();
            distributedLock=this.factory.createMultiLock(resourcePrefx,dataIds,expiryTime);
            if(log.isDebugEnabled()){
                long cost1=System.currentTimeMillis()-time1;
                log.debug("createMultiLock: "+cost1+"ms"+" currentThread: "+Thread.currentThread().getId()+" resource: "+mkId+"-"+dataCat+"-"+dataIds.size());
            }
            if (distributedLock == null){
                log.error("DistributedMultiLock get nothing!");
                result.setSuccess(false);
                return result;
                //调接口加锁是否成功是通过返回的结果进行判断，如果抛异常表明加锁失败，多数调接口的人并未捕获该异常，以为是发生了错误，所以此处还是直接返回false,并记录个日志，表明加不上锁是分布式锁未加上的标志
//                throw new DistributedLockException("pfcomm", DLErrorDefinition.AddLock_IsAcquired_Error,
//                        "DistributedLock get nothing!", null, ExceptionLevel.Error, false);
            }
            result=lockExtension.addBatchLock(mkId,dataCat,dataIds,groupId,persistenceTime,funcId,comment);
            return result;
        }catch (Throwable e){
            throw new io.iec.edp.caf.lock.service.api.exception.DistributedLockException("caf", DLErrorDefinition.AddBatchLock_Error,"addBatchLockError",e, ExceptionLevel.Error,false);
        }finally {
            try {
                if(distributedLock!=null){
                    distributedLock.multiUnlock();
                }
            } catch (Exception e) {
                log.error("multiunlock: ",e);
            }
        }
    }



    @Override
    @Deprecated
    public boolean removeLock(String lockId, boolean removeSharedLock, String sessionId) {
        DataValidator.checkForEmptyString(lockId, "lockId");
        return this.removeLock(lockId);
    }

    /**
     * 根据锁编号解锁
     *
     * @param lockId 锁编号
     * @return 是否成功的标识
     */
    @Override
    public boolean removeLock(String lockId) {
        return lockExtension.removeLock(lockId);
    }

    @Override
    public boolean removeLock(RemoveLockInstance removeLockInstance){
        return lockExtension.removeLock(removeLockInstance);
    }


    @Override
    public boolean removeSessionLock(String sessionId, int tenantId) {
        return lockExtension.removeSessionLock(sessionId,tenantId);
    }

    @Override
    public boolean removeBatchLock(String groupId) {
        return lockExtension.removeBatchLock(groupId);
    }

    @Override
    public  boolean removeBatchLockByContext(String contextId){
        return lockExtension.removeBatchLockByContext(contextId);
    }

    @Override
    public boolean removeInvalidLocks(int tenantId) {
        // to do
        return true;
    }




    /// <summary>
    /// 内部执行加锁逻辑
    /// </summary>
    private LockResult internalAddLock(String mkId, String dataCat, String dataId,
                                       DataLockOptions option, String funcId, String comment, String groupId, @Nullable DistributedLockOptions distributedLockOptions) {

        long timeIn = System.currentTimeMillis();

        Duration bizWaitTime=null;
        if(option!=null&&option.getTimeOut()!=null){
            bizWaitTime=option.getTimeOut();
        }
        long time=0;
        if(bizWaitTime!=null){
            time=bizWaitTime.toMillis();
        }
        long current=System.currentTimeMillis();
        LockResult result = new LockResult();
        //结果默认为不成功
        result.setSuccess(false);
        //根据上下文获取租户ID
        CAFContextInfo contextInfo=CAFContextResolverUtils.getCAFContextInfo();
        int tenantId =0;// this.contextService.getTenantId();
        if(contextInfo!=null){
            tenantId=contextInfo.getTenantId();
        }
        DistributedLock distributedLock = null;
        //根据资源、租户等信息构造锁编号
        // 获取持久资源
        String resource = String.format("%s-%s-%s-%s", tenantId, mkId, dataCat, dataId);

//        //如果不为实例级锁，首先检查session是否在线，不在线抛异常
//        if (this.isSessionNotValid(this.contextService.getSessionId())) {
//            throw new io.iec.edp.caf.lockservice.api.DistributedLockException("pfcomm", DLErrorDefinition.AddLock_IsSession_Error,
//                    "Session:" + this.contextService.getSessionId() + " is not online!", null, ExceptionLevel.Error, false);
//        }

//        LockEntity entity = this.createLockEntity(lockId, mkId, dataCat, dataId, option, funcId, comment,contextInfo);
        Duration expiryTime = distributedLockOptions == null ? Duration.ofSeconds(20) : distributedLockOptions.getExpiryTime();
        Duration disWaitTime=null;
        if(distributedLockOptions!=null&&distributedLockOptions.getWaitTime()!=null){
            disWaitTime=distributedLockOptions.getWaitTime();
        }
        try {
            if(disWaitTime==null){
                long time1 = System.currentTimeMillis();
                distributedLock = this.factory.createLock(resource, expiryTime);
                if(log.isDebugEnabled()){
                    long cost1=System.currentTimeMillis()-time1;
                    log.debug("createLock: "+cost1+"ms"+" currentThread: "+Thread.currentThread().getId()+" resource: "+mkId+"-"+dataCat+"-"+dataId);
                }
            }else{
                //支持加锁等待
                long time2 = System.currentTimeMillis();
                distributedLock=this.factory.tryCreateLock(disWaitTime,resource,expiryTime);
                if(log.isDebugEnabled()){
                    long cost2=System.currentTimeMillis()-time2;
                    log.debug("tryCreateLock: "+cost2+"ms"+" currentThread: "+Thread.currentThread().getId()+" resource: "+mkId+"-"+dataCat+"-"+dataId);
                }
            }
            //若得不到分布式锁则抛异常，因为如果还是返回false会混淆是gsplock表里有数据导致加不上锁还是redis里有数据导致加不上锁
            if (distributedLock == null || !distributedLock.isAcquired()){
                throw new DistributedLockException("pfcomm", DLErrorDefinition.AddLock_IsAcquired_Error,
                        "DistributedLock get nothing!", null, ExceptionLevel.Error, false);
            }
            result =lockExtension.addLock(mkId,dataCat,dataId,option,funcId,comment);//tryAddLock(mkId,dataCat,dataId,option,entity);
            if(time==0){  //未设置等待时长直接返回
                if(log.isDebugEnabled()){
                    long costTotal=System.currentTimeMillis()-timeIn;
                    log.debug("addLockTime: "+costTotal+"ms"+" currentThread: "+Thread.currentThread().getId()+" resource: "+mkId+"-"+dataCat+"-"+dataId);
                }
                return result;
            }else{
                if(result.isSuccess()){
                    if(log.isDebugEnabled()){
                        long costTotal=System.currentTimeMillis()-timeIn;
                        log.debug("lockTimeSuccess: "+costTotal+"ms"+" currentThread: "+Thread.currentThread().getId()+" resource: "+mkId+"-"+dataCat+"-"+dataId);
                    }
                    return result;      //加锁成功，直接返回
                }
                List<LockEntity> entities=result.getLockedEntities();

                time-=(System.currentTimeMillis()-current);
                //如果设置的业务锁等待时长waitTime已经超时了，则代表申请加锁失败返回
                if(time<=0){
                    result.setSuccess(false);
                    result.setLockedEntities(entities);
                    return result;
                }
                //未超时，跳入死循环尝试获取锁，然后根据加锁等结果进行跳出
                while(true){
                    long currentTime=System.currentTimeMillis();
                    result=lockExtension.addLock(mkId,dataCat,dataId,option,funcId,comment);//tryAddLock(mkId,dataCat,dataId,option,entity);
                    if(result.isSuccess()){
                        return result;
                    }
                    time-=(System.currentTimeMillis()-currentTime);
                    if(time<=0){
                        result.setSuccess(false);
                        result.setLockedEntities(entities);
                        return result;
                    }
                    if(time>900){
                        try{
                            Thread.sleep(900);
                        }catch (Exception e){
//                            throw new RuntimeException(e.getMessage());
                            log.error("thread sleep error: ",e);
                        }
                    }
                    time-=(System.currentTimeMillis()-currentTime);
                    if(time<=0){
                        result.setSuccess(false);
                        result.setLockedEntities(entities);
                        return result;
                    }
                }
            }

        } catch (Throwable e) {
            throw new io.iec.edp.caf.lock.service.api.exception.DistributedLockException("caf", DLErrorDefinition.AddLock_GetLock_Error,"addLockGetLockError",e, ExceptionLevel.Error,false);
        } finally {
            try {
                if (distributedLock != null){
                    distributedLock.close();
                }
            } catch (IOException e) {
                log.error("unlock: ",e);
//                throw new RuntimeException(e);
            }
        }
    }


    public LockStatus getLockStatus(LockEntity lockEntity) {
//        return checkValidate(lockEntity);
        return lockExtension.getLockStatus(lockEntity);
    }


    /**
     * 根据gropuId获取dataIds
     * @param groupId
     * @return
     */
    @Override
    public List<String>  getDataIds(String groupId){
        return  lockExtension.getDataIds(groupId);
    }


}
